#include "mq_channel.hpp"

namespace xzt
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;

        Connection(const VirtualHost::ptr& host, const ConsumerManager::ptr& p_cm, const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec, const threadpool::ptr& threadpool)
            : _host(host), _p_cm(p_cm), _conn(conn), _codec(codec), _threadpool(threadpool), _channel_manager(std::make_shared<ChannelManager>())
        {}

        ~Connection() = default;

        void openChannel(const openChannelRequestPtr& req)
        {
            bool ret = _channel_manager->openChannel(req->cid(), _host, _p_cm, _conn, _codec, _threadpool);
            if(ret == false)
            {
                _basicResponse(false, req->rid(), req->cid());
                return;
            }
            _basicResponse(true, req->rid(), req->cid());
        }

        void closeChannel(const closeChannelRequestPtr& req)
        {
            _channel_manager->closeChannel(req->cid());
            _basicResponse(true, req->rid(), req->cid());
        }

        Channel::ptr getChannel(const std::string& cid)
        {
            return _channel_manager->getChannel(cid);
        }

    private:
        void _basicResponse(bool is_success, const std::string& rid, const std::string& cid)
        {
            xzt::basicCommonReponse response;
            response.set_success(is_success);
            response.set_rid(rid);
            response.set_cid(cid);
            _codec->send(_conn, response);
        }

    private:
        VirtualHost::ptr _host;
        ConsumerManager::ptr _p_cm;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        threadpool::ptr _threadpool;
        ChannelManager::ptr _channel_manager;
    };

    // 连接管理器
    class ConnectionManager
    {
    public:
        using ptr = std::shared_ptr<ConnectionManager>;

        ConnectionManager() = default;
        ~ConnectionManager() = default;

        void newConnection(const VirtualHost::ptr& host, 
                            const ConsumerManager::ptr& p_cm, 
                            const muduo::net::TcpConnectionPtr& conn, 
                            const ProtobufCodecPtr& codec, 
                            const threadpool::ptr& threadpool)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto it = _connections.find(conn);
            if(it != _connections.end())    // 已经存在连接
            {
                return;
            }

            auto conn_ptr = std::make_shared<Connection>(host, p_cm, conn, codec, threadpool);
            _connections[conn] = conn_ptr;
        }

        void delConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            _connections.erase(conn);
        }

        Connection::ptr getConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            std::lock_guard<std::mutex> lock(_mutex);
            auto it = _connections.find(conn);
            if(it != _connections.end())    // 存在连接
            {
                return it->second;
            }
            return nullptr;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, Connection::ptr> _connections;  // 连接映射
    };
}

